package com.rabbit.producer.task;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.dataflow.DataflowJob;
import com.rabbit.producer.broker.RabbitBroker;
import com.rabbit.producer.constant.BrokerMessageStatus;
import com.rabbit.producer.entity.BrokerMessage;
import com.rabbit.producer.service.MessageStoreService;
import com.rabbit.task.annotation.ElasticJobConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * DataflowJob定时任务
 *
 * @author: 顾志杰
 */
@Component
@ElasticJobConfig(
        name = "com.rabbit.producer.task.RetryMessageDataflowJob",
        cron = "0/10 * * * * ?",
        description = "可靠性投递消息补偿任务",
        overwrite = true,
        shardingTotalCount = 1
)
@Slf4j
public class RetryMessageDataflowJob implements DataflowJob<BrokerMessage> {

    @Autowired
    private MessageStoreService messageStoreService;
    @Autowired
    private RabbitBroker rabbitBroker;
    /**
     * 最大重试次数
     */
    private static final int MAX_RETRY_COUNT = 3;

    @Override
    public List<BrokerMessage> fetchData(ShardingContext shardingContext) {
        List<BrokerMessage> brokerMessages = messageStoreService.fetchTimeOutMessageForRertry(BrokerMessageStatus.SENDING);
        log.info("----------------@@@@@@ 抓取数据集合,数量: {}   @@@@@@@@@---------", brokerMessages.size());
        return brokerMessages;
    }

    @Override
    public void processData(ShardingContext shardingContext, List<BrokerMessage> list) {
        list.forEach(brokerMessage -> {
            String messageId = brokerMessage.getMessageId();
            /*判断如果大于最大重试次数 就变为失败*/
            if (MAX_RETRY_COUNT <= brokerMessage.getTryCount()) {
                this.messageStoreService.failure(messageId);
                log.warn("---------消息重试最终失败，消息设置为最终失败，消息ID ：{}", messageId);
            } else {
                /*如果小于最大重试次数就继续投递*/
                /*重新发送需要更新tryCount*/
                this.messageStoreService.updateTryCount(messageId);
                this.rabbitBroker.reliantSend(brokerMessage.getMessage());

            }
        });
    }
}
